package cn.jltx.rxjava.rx.learn;

import android.util.Log;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

/**
 * @author jltxseo
 *         Created by junlintianxia on 2017/08/14.
 */

public class RxJava2Learn3 {

    /**
     * 不进行线程Scheduler切换控制时都是在当前主线程里面
     */
    public static void noScheduler() {
        Log.d(XqTag.TAG, "start.thread:" + Thread.currentThread());
        Observable.just("data")
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        Log.d(XqTag.TAG, "map0.thread:" + Thread.currentThread());
                        return s + "-map0";
                    }
                })
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(XqTag.TAG, "onSubscribe.thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onNext(String value) {
                        Log.d(XqTag.TAG, "onNext:" + value + ".thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(XqTag.TAG, "onError:" + e.toString() + ".thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onComplete() {
                        Log.d(XqTag.TAG, "onComplete.thread:" + Thread.currentThread());
                    }
                });
    }


    /**
     * 2、subscribeOn无论放置在哪个位置，它的线程切换发生在该Observable的OnSubscribe 中，即在它通知上一级 OnSubscribe 时，
     * 这时事件还没有开始发送，因此 subscribeOn() 的线程控制可以从事件发出的开端就造成影响。
     * 当使用了多个 subscribeOn() 的时候，只有第一个 subscribeOn() 起作用。
     */
    public static void subscribeOnScheduler() {
        Log.d(XqTag.TAG, "start.thread:" + Thread.currentThread());
        Observable.just("data")
                .subscribeOn(Schedulers.io())
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        Log.d(XqTag.TAG, "map0.thread:" + Thread.currentThread());
                        return s + "-map0";
                    }
                })
                .subscribeOn(Schedulers.newThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(XqTag.TAG, "onSubscribe.thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onNext(String value) {
                        Log.d(XqTag.TAG, "onNext:" + value + ".thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(XqTag.TAG, "onError:" + e.toString() + ".thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onComplete() {
                        Log.d(XqTag.TAG, "onComplete.thread:" + Thread.currentThread());
                    }
                });
    }

    /**
     * 3、observeOn的线程切换发生在自己内部的构建的Observable被订阅者中，也就是在它即将给下一级的订阅者发送事件时，所以ObserveOn()控制的是后面的线程，可以多个ObserveOn一起使用，
     * 每使用一次ObserveOn，它后面的线程就跟着变换一次。
     */
    public static void observeOnScheduler() {
        Log.d(XqTag.TAG, "start.thread:" + Thread.currentThread());
        Observable.just("data")
                .observeOn(Schedulers.io())
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        Log.d(XqTag.TAG, "map0.thread:" + Thread.currentThread());
                        return s + "-map0";
                    }
                })
                .observeOn(Schedulers.computation())
                .observeOn(Schedulers.newThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(XqTag.TAG, "onSubscribe.thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onNext(String value) {
                        Log.d(XqTag.TAG, "onNext:" + value + ".thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(XqTag.TAG, "onError:" + e.toString() + ".thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onComplete() {
                        Log.d(XqTag.TAG, "onComplete.thread:" + Thread.currentThread());
                    }
                });
    }

    /**
     * 4、下游的onSubscribe的回调是在执行subscribe()订阅动作的时候被调用了，因此不能先直接指定线程而是只能执行在subscribe()被调用时的线程中。
     * 5、对于设置doOnSubscribe回调默认跟onSubscribe的回调规则那样在执行subscribe()订阅动作的时候被调用了，但是如果在 doOnSubscribe() 之后有 subscribeOn() 的话，
     * 它将执行在离它最近的 subscribeOn() 所指定的线程。
     */
    public static void subscribeAndobserveOnScheduler() {
        Log.d(XqTag.TAG, "start.thread:" + Thread.currentThread());
        Observable.just("data")
                .observeOn(Schedulers.io())
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        Log.d(XqTag.TAG, "doOnSubscribe.thread:" + Thread.currentThread());
                    }
                })
                //doOnSubscribe后面设置了subscribeOn后执行离它最近的subscribeOn所指定的线程
                .subscribeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.io())
                .flatMap(new Function<String, ObservableSource<String>>() {
                    @Override
                    public ObservableSource<String> apply(String s) throws Exception {
                        Log.d(XqTag.TAG, "map0.thread:" + Thread.currentThread());
                        return Observable.just(s + "-map0");
                    }
                })
                .observeOn(Schedulers.computation())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(XqTag.TAG, "onSubscribe.thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onNext(String value) {
                        Log.d(XqTag.TAG, "onNext:" + value + ".thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(XqTag.TAG, "onError:" + e.toString() + ".thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onComplete() {
                        Log.d(XqTag.TAG, "onComplete.thread:" + Thread.currentThread());
                    }
                });
    }

    /**
     * subscribeOn和observeOn嵌套使用示例
     */
    public static void subscribeAndobserveOnNestScheduler() {
        Log.d(XqTag.TAG, "start.thread:" + Thread.currentThread());
        Observable.just("data")
                .observeOn(Schedulers.io())
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        Log.d(XqTag.TAG, "doOnSubscribe.thread:" + Thread.currentThread());
                    }
                })
                //doOnSubscribe后面设置了subscribeOn后执行离它最近的subscribeOn所指定的线程
                .subscribeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.io())
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        Log.d(XqTag.TAG, "map0.thread:" + Thread.currentThread());
                        return s + "-map0";
                    }
                })
                .observeOn(Schedulers.newThread())
                .flatMap(new Function<String, ObservableSource<String>>() {
                    @Override
                    public ObservableSource<String> apply(String s) throws Exception {
                        Log.d(XqTag.TAG, "flatMap1.thread:" + Thread.currentThread());
                        Observable<String> flatMapObservable = Observable.just(s + "-flatMap1")
                                .doOnSubscribe(new Consumer<Disposable>() {
                                    @Override
                                    public void accept(Disposable disposable) throws Exception {
                                        Log.d(XqTag.TAG, "flatMapObservable.doOnSubscribe.thread:" + Thread.currentThread());
                                    }
                                })
                                .subscribeOn(AndroidSchedulers.mainThread())
                                .observeOn(Schedulers.io())
                                .map(new Function<String, String>() {
                                    @Override
                                    public String apply(String s) throws Exception {
                                        Log.d(XqTag.TAG, "apply.map2.thread:" + Thread.currentThread());
                                        return s + "-map2";
                                    }
                                })
                                .observeOn(Schedulers.single())
                                .doOnNext(new Consumer<String>() {
                                    @Override
                                    public void accept(String s) throws Exception {
                                        Log.d(XqTag.TAG, "doOnNext.accept" + s + ".map2.thread:" + Thread.currentThread());
                                    }
                                })
                                .observeOn(Schedulers.newThread())
                                .doOnComplete(new Action() {
                                    @Override
                                    public void run() throws Exception {
                                        Log.d(XqTag.TAG, "doOnComplete.accept" + s + ".map2.thread:" + Thread.currentThread());
                                    }
                                });
                        return flatMapObservable;
                    }
                })
                .observeOn(Schedulers.computation())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(XqTag.TAG, "onSubscribe.thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onNext(String value) {
                        Log.d(XqTag.TAG, "onNext:" + value + ".thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(XqTag.TAG, "onError:" + e.toString() + ".thread:" + Thread.currentThread());
                    }

                    @Override
                    public void onComplete() {
                        Log.d(XqTag.TAG, "onComplete.thread:" + Thread.currentThread());
                    }
                });
    }

}
